package com.liuhm.core;

import com.google.common.collect.Lists;
import com.univocity.parsers.csv.CsvParser;
import com.univocity.parsers.csv.CsvParserSettings;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.io.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicLong;

/**
 *  easyExcel 取数excel文件数据
 * @author liuhaomin
 * @date 2020/10/30
 */
@Slf4j
public class CsvBroker extends AbstractBroker {

	private AtomicLong count = new AtomicLong(0);


	@Override
	public void producer(File file,  int startRow, int columnSize) throws Exception {
		InputStream inputStream = new BufferedInputStream(new FileInputStream(file));
		LinkedBlockingQueue<List<String>> queue = getQueue();
		String ext = file.getName().substring(file.getName().lastIndexOf("."));
		if(".csv".equals(ext) ){
			importReadCsv(file, startRow, columnSize, inputStream, queue);
		}else{
			log.error("不能导入{}结尾的",ext);
		}
	}

	/**
	 * 导入读取excel
	 * @param
	 * @return
	 */
	/*private void importReadExcel(File file, int startRow, int columnSize, int startSheet, List<Integer> sheets, InputStream inputStream, LinkedBlockingQueue<List<String>> queue) {
		log.info("========================读线程开始========================");
		ExcelReader excelReader = null;
		try {
			excelReader = EasyExcel.read(inputStream,
					new AnalysisEventListener<List<String>>() {
						@Override
						public void invoke(List<String> object, AnalysisContext context) throws ExcelAnalysisStopException {
							// 根据key获取当前导入是否取消
							if (isCancel){
								throw new ExcelAnalysisStopException("用户取消导入，停止解析excel(忽略该异常，用户中断导入抛出)");
							}

							// 添加数据到队列
							// 当前 sheet 下标编号（从 0 开始）
							ReadSheetHolder readSheetHolder = context.readSheetHolder();
							Integer sheetNo = readSheetHolder.getSheetNo();
							// 当前行索引
							ReadRowHolder readRowHolder = context.readRowHolder();
							Integer row = readRowHolder.getRowIndex();
							int listSize = object.size();

							List<String> tempList = Lists.newArrayList();
							if (listSize > columnSize){
								// 超过表头数据量的截取
								tempList = object.subList(0, columnSize);
							}else {
								// 不足的补充null
								tempList.addAll(object);
								for (int i = listSize; i < columnSize; i++){
									tempList.add(null);
								}
							}
							try {
								if(sheetNo == startSheet && row >= startRow){
									// 表头下一行数据开始添加到队列
									if(!checkListIsNullString(tempList)){
										queue.put(tempList);
										count.incrementAndGet();
									}
								}else if(sheetNo != startSheet){
									// 其他sheet数据直接添加到队列
									if(!checkListIsNullString(tempList)){
										queue.put(tempList);
										count.incrementAndGet();
									}
								}
							} catch (InterruptedException e) {
								log.error("========================导入数据文件-读取数据放入队列异常:{}===================\n" + e);
							}
						}


						@Override
						public void doAfterAllAnalysed(AnalysisContext context) {

						}
					}).registerConverter(new CustomStringNumberConverter()).setDefaultReturnMap(Boolean.FALSE).build();

			// 读取 excel 文件 sheet 数量
			List<ReadSheet> readSheets = excelReader.excelExecutor().sheetList();
			ReadSheet readSheet;

			//读取全部sheet
			if (sheets == null || sheets.size() == 0) {
				log.info("正在读取 sheet = {}", startSheet + 1);
				//指定的开始sheet
				readSheet = EasyExcel.readSheet(startSheet).headRowNumber(startRow -1).build();
				excelReader.read(readSheet);
				for (int i = 0; i < readSheets.size(); i++) {
					if (i != startSheet) {
						log.info("正在读取 sheet = {}", i + 1);
						readSheet = EasyExcel.readSheet(i).headRowNumber(0).build();
						excelReader.read(readSheet);
					}
				}
			}else {
				//勾选的sheet
				if (sheets.contains(startSheet)) {
					log.info("正在读取 sheet = {}", startSheet + 1);
					//指定的开始sheet
					readSheet = EasyExcel.readSheet(startSheet).headRowNumber(startRow -1).build();
					excelReader.read(readSheet);
				}
				for (int i = 0; i < readSheets.size(); i++) {
					if (sheets.contains(i) && i != startSheet) {
						log.info("正在读取 sheet = {}", i + 1);
						readSheet = EasyExcel.readSheet(i).headRowNumber(0).build();
						excelReader.read(readSheet);
					}
				}
			}

			// 读取完成，修改文件状态
			isReadFinished = true;
			log.info("========================读线程完成========================;\n "
					+ "isReadFinished = {}，读取数据量 count = {}", isReadFinished, count.get());
		} catch (Exception e) {
			log.info("导入数据文件[{}]读数据过程中出现异常：{}", file.getName(), e.getMessage());
			// 设置读过程中发生异常
			setExceptionFinished(true);
			isReadFinished = true;
			log.error("========================读线程完成(异常结束)========================;\n "
					+ "isReadFinished = {}，isExceptionFinished={}，读取数据量 count = {}", isReadFinished, isExceptionFinished(), count.get());
		} finally {
			try {
				if(excelReader != null){
					excelReader.finish();
				}
				inputStream.close();
			} catch (IOException e) {
				log.error("========================读线程完成关闭流异常========================;\n {}", e);
			}
		}
	}
	*/
	/**
	 * 导入读取excel
	 * @param
	 * @return
	 */
	private void importReadCsv(File file, int startRow, int columnSize,  InputStream inputStream, LinkedBlockingQueue<List<String>> queue) {
		log.info("========================读线程开始========================");
		// 根据key获取当前导入是否取消
		String cutCvsPath = file.getAbsolutePath().substring(0,file.getAbsolutePath().lastIndexOf("."));
		File cutCvsFile = new File(cutCvsPath);
		List<File> fileList = new ArrayList<>();
		// 是否有子文件
		boolean isCut = false;
		if (!cutCvsFile.isDirectory()) {
			isCut = false;
		} else if (cutCvsFile.isDirectory()) {
			String[] files = cutCvsFile.list();
			for (int i = 0; i < files.length; i++) {
				if(files[i].indexOf(".csv")<0){
					fileList.add(new File(cutCvsPath+File.separator+files[i]));
				}
			}
			log.info("文件夹个数{}",files.length);
			isCut = true;
		}

		try {
			if(!isCut){
				CvsPutQueue(startRow, columnSize, file, queue);
			}else{
				log.info("getCsvReadThreadNum{}",getCsvReadThreadNum());
				ThreadPoolExecutor exec = (ThreadPoolExecutor) Executors.newFixedThreadPool(getCsvReadThreadNum());
				for(int i = 0;i<fileList.size();i++) {
					exec.submit(new MyThread(fileList.get(i),columnSize,queue,count,startRow));
				}
				exec.shutdown();
			}
			log.info("========================读线程完成========================;\n "
					+ "读取数据量 count = {}", count.get());
		} catch (Exception e) {
			log.info("导入数据文件[{}]读数据过程中出现异常：{},{}", file.getName(), e.getMessage(),e);
			log.error("========================读线程完成(异常结束)========================;\n "
					+ "读取数据量 count = {}", count.get());
		}
	}

	static class MyThread extends Thread{
		private File file;
		private int columnSize;
		private int startRow;
		private AtomicLong count;
		private LinkedBlockingQueue<List<String>> queue;
		public MyThread(File file, int columnSize,LinkedBlockingQueue<List<String>> queue,AtomicLong count,int startRow)
		{
			this.file = file;
			this.columnSize = columnSize;
			this.queue = queue;
			this.count = count;
			this.startRow = startRow;
		}

		@Override
		public void run() {
			Long startTime=System.currentTimeMillis();
			int readCount=0;
			try {
				CsvParserSettings settings = new CsvParserSettings();
				settings.getFormat().setLineSeparator("\n");
				CsvParser csvParser = new CsvParser(settings);
				csvParser.beginParsing(file,"UTF-8");
				String[] strings;
				while ((strings = csvParser.parseNext()) != null) {
					// 当前行索引
					Long row =csvParser.getContext().currentLine();
					List<String> object= Arrays.asList(strings);
					int listSize = object.size();
					List<String> tempList = Lists.newArrayList();
					if (listSize > columnSize){
						// 超过表头数据量的截取
						tempList = object.subList(0, columnSize);
					}else {
						// 不足的补充null
						tempList.addAll(object);
						for (int i = listSize; i < columnSize; i++){
							tempList.add(null);
						}
					}
					// test_aa的文件是有头的文件
					if( row== startRow&&file.getName().equals("test_aa")){
						continue;
					}
					// 表头下一行数据开始添加到队列
					if(!checkListIsNullString(tempList)){
						queue.put(tempList);
						count.incrementAndGet();
						readCount++;

					}
				}
				csvParser.stopParsing();
			} catch (InterruptedException e) {
				log.error("========================导入数据文件-读取数据放入队列异常:{}===================\n" + e);
			}catch (Exception e) {
				log.error("========================导入数据文件-读取数据放入队列异常:{}===================\n" + e);
			}

			log.info("读----------{},读取到队列花费{},读取条数{} " ,Thread.currentThread().getName(),(System.currentTimeMillis()-startTime),readCount);
		}
	}
	/**
	 * 没有切割读取cvs文件值放入队列
	 * @param
	 * @return
	 */
	private void CvsPutQueue(int startRow, int columnSize, File file, LinkedBlockingQueue<List<String>> queue) {
		CsvParserSettings settings = new CsvParserSettings();
		settings.getFormat().setLineSeparator("\n");
		CsvParser csvParser = new CsvParser(settings);
		csvParser.beginParsing(file,"UTF-8");
		String[] strings;
		while ((strings = csvParser.parseNext()) != null) {
			// 当前行索引
			Long row =csvParser.getContext().currentLine();
			List<String> object= Arrays.asList(strings);
			int listSize = object.size();
			List<String> tempList = Lists.newArrayList();
			if (listSize > columnSize){
				// 超过表头数据量的截取
				tempList = object.subList(0, columnSize);
			}else {
				// 不足的补充null
				tempList.addAll(object);
				for (int i = listSize; i < columnSize; i++){
					tempList.add(null);
				}
			}
			try {
				if( row > startRow){
					// 表头下一行数据开始添加到队列
					if(!checkListIsNullString(tempList)){
						queue.put(tempList);
						count.incrementAndGet();
					}
				}
			} catch (InterruptedException e) {
				log.error("========================导入数据文件-读取数据放入队列异常:{}===================\n" + e);
			}
		}
		csvParser.stopParsing();
	}

	private static boolean checkListIsNullString(List<String> list){
		if(null != list && list.size() > 0){
			for (String s : list) {
				if(!StringUtils.isEmpty(s)){
					return false;
				}
			}
		}
		return true;
	}

}
